package pbc.gx.SaveJson.kafka.kafka2redis;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.context.ApplicationContext;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.CrossOrigin;
import pbc.gx.SaveJson.kafka.service.SaveJsonService;

import java.util.*;

@Slf4j
public class Save2RedisThread extends  Thread {
    private final RedisTemplate redisTemplate;
    private final ApplicationContext applicationContext = ApplicationContextProvider.getApplicationContext();

    private final List<ConsumerRecord<String, JSONObject>> records;

    public Save2RedisThread(RedisTemplate redisTemplate, List<ConsumerRecord<String, JSONObject>> records) {
        this.redisTemplate = redisTemplate;
        this.records = records;
    }


    public void run() {
        SaveJsonService saveJsonService = applicationContext.getBean(SaveJsonService.class);
        saveJsonService.save2Redis(redisTemplate, records);
        //log.info("Redis:" + this.getId() + " " + records.size());
    }
}
